/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */

/* 
 * File:   KKSFile.cpp
 * Author: ever
 * 
 * Created on 2018年5月9日, 上午11:03
 */

#include <vector>

#include "KKSFile.h"

KKSFile::KKSFile(const string &file) {
    attach(file);
}

KKSFile::~KKSFile() {
    if (_fd > 0) close(_fd);

    auto it = _kks_meta_sets.begin();
    while (it != _kks_meta_sets.end()) {
        delete it->second;
        ++it;
    }
    _kks_meta_sets.clear();
}

void KKSFile::attach(const string &file) {
    _mmap.mmap(file.c_str(), KKS_FILE_HEAD_MAX_SIZE);
    _fd = open(file.c_str(), O_EXCL | O_RDWR, 0666);
    _file_length = TL_Common::getFileLength(_fd);
    char *addr = (char *) _mmap.getPointer();
    _head = (KKS_File_Head *) addr;
    _meta = (KKS_Meta *) (addr + sizeof(KKS_File_Head));
    _kfile_name = file;

    if (_head->magic != KKS_FILE_MAGIC) {
        //全新文件
        *_head = KKS_File_Head();
        memset(_meta, 0, sizeof(KKS_Meta) * KKS_MAX_COUNT_PEER_FILE);
    } else {
        KKS_Meta *meta_tmp = _meta;
        for (u32_t i = 0; i < _head->kks_count; ++i) {
            _all_kks.insert(meta_tmp->kks_id);
            auto r = _kks_meta_sets.insert(make_pair(meta_tmp, new KKS_Meta_Value()));
            ++meta_tmp;
        }
        _all_kks.erase(0);
    }

}

const string &KKSFile::getKKSFileName() {
    return _kfile_name;
}

KKSFile::KKS_File_Head *KKSFile::getKKSFileHead() {
    return _head;
}

KKSFile::KKS_Meta *KKSFile::getMeta(u32_t id) {
    if (id == 0) return _meta;
    KKS_Meta meta;
    meta.kks_id = id;
    auto it = _kks_meta_sets.find(&meta);
    if (it != _kks_meta_sets.end()) {
        return it->first;
    }
    return nullptr;
}

void KKSFile::getRowOffset(u32_t id, time_t stime, time_t etime, vector<off_t> &rows) {
    rows.clear();
    KKS_Meta *meta = getMeta(id);
    if (!meta) return;
    KKS_ROW_Head rhead;
    off_t row_off = meta->offset_first;
    while (row_off > 0 && row_off <= meta->offset_current) {
        pread(_fd, &rhead, sizeof(KKS_ROW_Head), row_off);
        if (stime <= rhead.tail_tm && etime >= rhead.start_tm)
            rows.push_back(row_off);

        //tm是递增的，所以可以提前结束检查
        if (rhead.start_tm > etime) break;
        row_off = rhead.offset_next;
    }
}

vector<off_t> KKSFile::getRowOffset(u32_t id) {
    vector<off_t> rows;
    KKS_Meta *meta = getMeta(id);
    if (!meta) return rows;
    KKS_ROW_Head rhead;
    off_t row_off = meta->offset_first;
    rows.push_back(row_off);
    while (row_off > 0 && row_off <= meta->offset_current) {
        pread(_fd, &rhead, sizeof(KKS_ROW_Head), row_off);
        row_off = rhead.offset_next;
    }
    return rows;
}


bool KKSFile::getRowData(off_t offset, vector<char> &buffer) {
    buffer.clear();
    buffer.reserve(KSS_ROW_LENGTH);
    char *addr = &buffer[0];
    ssize_t len = pread(_fd, addr, KSS_ROW_LENGTH, offset);

    return len == KSS_ROW_LENGTH;
}

KKSFile::KKS_Meta_Value *KKSFile::getMetaValue(u32_t id) {
    KKS_Meta meta;
    meta.kks_id = id;
    auto it = _kks_meta_sets.find(&meta);
    if (it != _kks_meta_sets.end()) {

        return it->second;
    }
    return nullptr;
}

void KKSFile::addKKS(u32_t id, u32_t tm, u32_t vl, char quality) {
    if (_max_count > 0) {
        reduceRowsByCount(_max_count);
    }

    KKS_Meta meta;
    meta.kks_id = id;
    auto it = _kks_meta_sets.find(&meta);
    if (it != _kks_meta_sets.end()) {
        KKS_Meta_Value *v = it->second;
        //使用的时候加载
        if (v->row_head == nullptr)
            lazyLoadRow(it->first, v);
        KKS_ROW_Head *row_head = v->row_head;
        if (row_head->count == KKS_ROW_COUNT) {
            //上一个节点已经满了，直接分配新row来记录值
            allocRow(it->first, v, tm, vl, quality);
            return;
        }

        KKS_Value *value = v->row_tail_value;
        //todo 时间值重复
        //if(tm == v->row_head->tail_tm){}
        //int tail_delta_delta = getDeltaDelatValue(row_head, value - 1);
        int delta = tm - row_head->tail_tm;
        int new_delta_delta = delta - row_head->delta; //tm - row_head->tail_tm - tail_delta_delta;
        if (new_delta_delta > 0x3F || new_delta_delta < -0x3F) {
            //这里至少需要2个空间
            if (row_head->count + 1 >= KKS_ROW_COUNT) {
                //不满足要求，直接放到下一个row中
                allocRow(it->first, v, tm, vl, quality);
                return;
            } else {
                KKS_Value *value_next = value + 1;
                value->delta_delta_tm = 0x80;
                value->value = tm;
                value_next->delta_delta_tm = 0;
                value_next->value = vl;
                value_next->quality = quality;
                row_head->delta = 0;
                row_head->tail_tm = tm;
                row_head->count += 2;
                v->row_tail_value += 2;
            }
        } else {
            bool is_negative = new_delta_delta < 0;
            if (is_negative) {
                new_delta_delta = -new_delta_delta;
                new_delta_delta |= 0x40;
            }

            value->delta_delta_tm = new_delta_delta;
            value->value = vl;
            value->quality = quality;
            row_head->delta = delta;
            row_head->tail_tm = tm;

            ++row_head->count;
            ++v->row_tail_value;
        }
    } else {
        //new KKS

        allocKKS(id, tm, vl, quality);
    }
}

/*
int KKSFile::getDeltaDelatValue(KKSFile::KKS_ROW_Head* head, KKSFile::KKS_Value* value) {
    int r = 0;
    if (value->delta_delta_tm & 0x80) {
        memcpy(&r, &value->value, sizeof (r));
        return r;
        //warning: may result in an unaligned pointer value [-Waddress-of-packed-member]
        //return *((int *) (&(value->value)));
    } else {
        r = value->delta_delta_tm & 0x3F;
        if (value->delta_delta_tm & 0x40) {
            return -r;
        } else {
            return r;
        }
    }
}
 */

void KKSFile::allocKKS(u32_t id, u32_t tm, u32_t vl, char quality) {

    KKS_Meta *new_meta = _meta + _head->kks_count;
    *new_meta = KKS_Meta(); //init
    new_meta->kks_id = id;
    new_meta->offset_first = _head->offset_tail;
    //new_meta->offset_current = _head->offset_tail;

    //分配row数据
    allocRow(new_meta, NULL, tm, vl, quality);
    ++_head->kks_count;
}

//分配row，并记录第一个数据

KKSFile::KKS_Meta_Value *KKSFile::allocRow(KKS_Meta *k, KKS_Meta_Value *v, u32_t tm, u32_t vl, char quality, bool delete_old_v) {
    //KKSFile::KKS_Meta_Value * KKSFile::allocRow(KKS_Meta * k, KKS_Meta_Value * v, bool delete_old_v) {
    KKS_Meta_Value *nv = new KKS_Meta_Value();
    if ((_head->offset_tail + KSS_ROW_LENGTH) > _file_length) {
        extendFileSize(_head->offset_tail + KSS_ROW_LENGTH * 10);
    }
    //cerr << "mmap:" << __LINE__ << "," << KSS_ROW_LENGTH << "," << _fd << "," << _head->offset_tail << endl;
    nv->memap.mmap(KSS_ROW_LENGTH, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, _head->offset_tail);
    char *addr = (char *) nv->memap.getPointer();
    nv->row_head = (KKS_ROW_Head *) addr;
    //* nv->row_head = KKS_ROW_Head(); //初始值
    nv->row_tail_value = (KKS_Value *) (addr + sizeof(KKS_ROW_Head));

    nv->row_head->start_tm = nv->row_head->tail_tm = tm;
    nv->row_head->offset_next = 0;
    nv->row_head->count = 1;
    nv->row_tail_value->delta_delta_tm = 0;
    nv->row_tail_value->value = vl;
    nv->row_tail_value->quality = quality;
    nv->row_head->count = 1;
    ++nv->row_tail_value;
    //修改meta，为reload最快速导航，避免查找链表
    k->offset_current = _head->offset_tail;

    if (v) {
        //链接offset；如果v==NULL，通常是分配首个row；否则会导致链表丢失
        v->row_head->offset_next = _head->offset_tail;
        if (delete_old_v) delete v;
        --_row_count;
    }
    _kks_meta_sets[k] = nv;
    _head->offset_tail += KSS_ROW_LENGTH;

    ++_row_count;
    return nv;
}

KKSFile::KKS_Meta_Value *KKSFile::lazyLoadRow(KKS_Meta *k, KKS_Meta_Value *v) {
    if (v->row_head == nullptr) {
        //cerr << "mmap:" << __LINE__ << "," << KSS_ROW_LENGTH << "," << _fd << "," << _head->offset_tail << endl;

        v->memap.mmap(KSS_ROW_LENGTH, PROT_READ | PROT_WRITE, MAP_SHARED, _fd, k->offset_current);
        char *addr = (char *) v->memap.getPointer();
        v->row_head = (KKS_ROW_Head *) addr;
        v->row_tail_value = (KKS_Value *) (addr + sizeof(KKS_ROW_Head) + sizeof(KKS_Value) * (v->row_head->count));
        ++_row_count;
    }
    return v;
}

void KKSFile::extendFileSize(off_t length) {
    if (_file_length >= length) return;

    if (length < 1) return;
    pwrite(_fd, "\0", 1, length - 1);
    _file_length = length;
}

void KKSFile::reduceRowsByCount(u32_t count) {
    if (_row_count < count) return;
    std::map<time_t, std::vector<KKS_Meta_Value *> > _tmsort_val;
    for (auto it : _kks_meta_sets) {
        KKS_ROW_Head *row_head = it.second->row_head;
        if (row_head) {
            _tmsort_val[row_head->tail_tm].push_back(it.second);
        }
    }
    while (_row_count > count) {
        auto vkmv = _tmsort_val.begin()->second;
        for (auto &it : vkmv) {
            KKS_Meta_Value *kmv = it;
            kmv->memap.munmap();
            kmv->row_head = nullptr;
            kmv->row_tail_value = nullptr;
        }
        _row_count -= vkmv.size();
        _tmsort_val.erase(_tmsort_val.begin());
    }
}

void KKSFile::setMaxMemUsage(u64_t mem) {
    u32_t count = mem / KSS_ROW_LENGTH;
    if (count > 0) {
        if (count < 20000) count = 20000;
        _max_count = count;
        reduceRowsByCount(_max_count);
    }
}

void KKSFile::test() {
    auto it = _kks_meta_sets.begin();
    while (it != _kks_meta_sets.end()) {
        KKS_Meta *m = it->first;
        //KKS_Meta_Value* mv = it->second;
        cerr << "kdd_id:" << m->kks_id << ",first:" << m->offset_first << ",current:" << m->offset_current << endl;
        KKS_ROW_Head rhead;
        off_t row_off = m->offset_first;
        while (row_off > 0 && row_off <= m->offset_current) {
            pread(_fd, &rhead, sizeof(KKS_ROW_Head), row_off);
            row_off = rhead.offset_next;
            cerr << "next off_t:" << row_off << ",tm:[" << rhead.start_tm << "-" << rhead.tail_tm << "],count:" << rhead.count << endl;
        }
        ++it;
    }
}
